#include "RpcConnection.h"
#include "ConnectionListener.h"
#include "RpcConnectionListener.h"



RpcConnection::RpcConnection(boost::asio::io_service& io_service, 
							 ConnectionClosedListener* listerner
							 ):
	Connection(io_service, listerner),
	m_breathe_timer(30000),
	m_breather_sendcount(0),
	m_breather_recvcount(0)

{
}

RpcConnection::~RpcConnection(void)
{
}

bool RpcConnection::onReadRawData(uint8* buffer, size_t s)
{
	LogFunctionEnter;

	//LOG_INFO << "onReadRawData " << s;

 	package_parser_.parse(buffer, buffer + s,
 		boost::bind(&RpcConnection::onHandlePackage,this,_1,_2));

	return true;
}

void RpcConnection::breathe()
{
	LogFunctionEnter;

	++m_breather_sendcount;

	int unSynWindow = abs(m_breather_sendcount - m_breather_recvcount);
	//VLOG(10) << "breathe unSysnWindow: " << unSynWindow;


	if (unSynWindow > 2)
	{
		LOG_WARNING << "breathe failed width sendcout: " 
			<< m_breather_sendcount << "recvcout: " << m_breather_recvcount;
		stop();
		return;
	}

	ArrayBuffer* buffer = new ArrayBuffer(sizeof (PacketHead));
	PacketHead phd(sizeof PacketHead,CMD_BREATHE);
	(*buffer) << phd;

	pushData(buffer);
	startWrite();
}

void RpcConnection::update(uint32 timeElapse)
{
	Connection::update(timeElapse);

	if (isStopped())
		return;

	m_breathe_timer.update(timeElapse);
	if (m_breathe_timer.passed())
	{
		m_breathe_timer.reset();
		breathe();
	}
}

static
void onCallServiceMethodCallback(ConnectionPtr c, uint32 method_id,
								  ::google::protobuf::Message* request,
								  ::google::protobuf::Message* response)
{
	//VLOG(4) << "onCallMethodCallback: " << method_name;

	delete request;

	ArrayBuffer* buffer_hd = new ArrayBuffer(sizeof (PacketHead));
	PacketHead phd(sizeof PacketHead,CMD_RESPONE);
	(*buffer_hd) << phd;

	int bs = response->ByteSize();
	ArrayBuffer* buffer = new ArrayBuffer(bs);
	uint8* tmpbs = buffer->gain(bs);

	if (!response->SerializeToArray(tmpbs,bs))
	{
		delete response;
		delete buffer;
		return;
	}

	delete response;

	c->pushData(buffer_hd);
	c->pushData(buffer);
	c->startWrite();
}

void RpcConnection::CallServiceMethod(google::protobuf::Service* service, 
									  uint32 method_id, ArrayBuffer& buffer)
{
	const ::google::protobuf::MethodDescriptor* md = 
		service->GetDescriptor()->method(method_id);
	if (md)
	{
		::google::protobuf::Message* req = m_service->GetRequestPrototype(md).New();
		::google::protobuf::Message* res = m_service->GetResponsePrototype(md).New();

		if (!req->ParseFromArray(buffer.data(),buffer.size()))
		{
			delete req;
			delete res;
			return;
		}

		RPCClosure* closure = RPCClosure::New(
			boost::bind(onCallServiceMethodCallback,shared_from_this(),method_id,req,res));

		LOG_INFO << "Call Service Method " <<  "method_id: " << method_id ;
		service->CallMethod(md,this,req,res,closure);
	}
}

void RpcConnection::onHandlePackage(PacketHead& pkh, ArrayBuffer& buffer)
{
	// 	if (isClosed())
	// 		return false;

	LogFunctionEnter;

	if (pkh.cmd == CMD_PING)
	{
		m_breather_recvcount++;
	}
	else if (pkh.cmd == CMD_REQUEST)
	{
		RequestHead rh;
		try
		{
			buffer >> rh;
		}
		catch (...)
		{
			return;
		}
		
		onHandleRequest(rh,buffer);
	}
	else if (pkh.cmd == CMD_RESPONE)
	{	
		ResponseHead rh;
		try
		{
			buffer >> rh;
		}
		catch (...)
		{
			return;
		}

		ResponseMap::iterator it = m_response_handler.find(rh.method);
		if (it != m_response_handler.end())
		{
			ResponseData respdat = it->second;
			if (!respdat.response->ParseFromArray(buffer.data(), buffer.size()))
			{
				respdat.controller->SetFailed("Response ParseFromArray failed");
			}
			
			respdat.done->Run();

			m_response_handler.erase(it);
		}
	}
	else
	{
		ASSERT(false);
	}
	//VLOG(10) << "handle package " << pk.getDataSize() << " ";

// 	RPC::Package rpc_pk;
// 	if (rpc_pk.ParseFromArray(buffer.data(),buffer.size()))
// 	{
// 		if (RPC::Package_Type_BREATHE == rpc_pk.type())
// 		{
// 			//VLOG(10) << "recv package Package_Type_BREATHE";
// 
// 			m_breather_recvcount++;
// 		}
// 		else if (rpc_pk.type() == RPC::Package::REQUEST)
// 		{
// 			//VLOG(4) << "recv package REQUEST";
// 
// 			if (m_service.get() == NULL)
// 				return;
// 
// 			const ::google::protobuf::MethodDescriptor* md = m_service->GetDescriptor()->FindMethodByName(rpc_pk.method());
// 			if (md)
// 			{
// 				::google::protobuf::Message* req = m_service->GetRequestPrototype(md).New();
// 				::google::protobuf::Message* res = m_service->GetResponsePrototype(md).New();
// 
// 				if (!req->ParseFromString(rpc_pk.request_content()))
// 				{
// 					delete req;
// 					delete res;
// 					return;
// 				}
// 
// 				if (!res->ParseFromString(rpc_pk.response_content()))
// 				{
// 					delete req;
// 					delete res;
// 					return;
// 				}
// 
// 				RPCClosure* closure = RPCClosure::New(
// 					boost::bind(onCallMethodCallback,shared_from_this(),rpc_pk.method(),req,res));
// 
// 				LOG_INFO << "CallServiceMethod " <<  "method_name: " << md->name() ;
// 				m_service->CallMethod(md,this,req,res,closure);
// 			}
// 		}
// 		else if(rpc_pk.type() == RPC::Package::RESPONSE)
// 		{
// //			VLOG(4) << "recv package RESPONSE";
// 
// 			if (m_service.get() == NULL)
// 				return;
// 
// 			const ::google::protobuf::MethodDescriptor* md = m_service->GetDescriptor()->FindMethodByName(rpc_pk.method());
// 			if (md)
// 			{
// 				ResponseMap::iterator it = m_response_handler.find(
// 					m_service->GetResponsePrototype(md).GetDescriptor()->full_name());
// 				if (it != m_response_handler.end())
// 				{
// 					ResponseData respdat = it->second;
// 					respdat.response->ParseFromString(rpc_pk.response_content());
// 					respdat.done->Run();
// 
// 					m_response_handler.erase(it);
// 				}
// 			}
// 
// 		}
// 		else
// 		{
// 			assert(false);
// 		}
// 	}

	return;
}

void RpcConnection::CallMethod(const google::protobuf::MethodDescriptor *method,
							google::protobuf::RpcController *controller,
							const google::protobuf::Message *request,
							google::protobuf::Message *response,
							google::protobuf::Closure *done)
{

	const int method_id = method->index();
	//const std::string& requst_name = request->GetDescriptor()->full_name();
	const std::string& response_name = response->GetDescriptor()->full_name();
	//RPC::Package pack;

	LOG_INFO << "CallMethod " <<  "method_id: " << method_id ;
	LOG_INFO << "response_name: " << response_name;


	ArrayBuffer* buffer_hd = new ArrayBuffer(sizeof (PacketHead));
	PacketHead phd(sizeof PacketHead,CMD_REQUEST);
	(*buffer_hd) << phd;
	

	int bs = request->ByteSize();
	ArrayBuffer* buffer = new ArrayBuffer(bs);
	uint8* tmpbs = buffer->gain(bs);

	if (!request->SerializeToArray(tmpbs,bs))
	{
		delete buffer;

		controller->SetFailed("request SerializeToArray failed!");
		done->Run();
		return;
	}

	pushData(buffer_hd);
	pushData(buffer);
	startWrite();

// 	if (NULL == response)
// 	{
// 		done->Run();
// 		return;
// 	}
 
 	m_response_handler.insert(
		std::make_pair(method_id,ResponseData(controller,request,response,done)));
}

// std::string* RpcConnection::encodePack(RPC::Package::Type t , const std::string& method, 
// 									const google::protobuf::Message * request, 
// 									const google::protobuf::Message * response)
// {
// 	std::string content;
// 
// 	RPC::Package pack;
// 	pack.set_type(t);
// 	if (!method.empty())
// 		pack.set_method(method);
// 	else
// 		return NULL;
// 
// 	if (request)
// 	{
// 		if (!request->SerializeToString(&content))
// 			return NULL;
// 		pack.set_request_content(content);
// 	}
// 
// 	if (response)
// 	{
// 		if (!response->SerializeToString(&content))
// 			return NULL;
// 		pack.set_response_content(content);
// 	}
// 
// 	std::string* pack_string = new std::string;
// 	if (!pack.SerializeToString(pack_string))
// 	{
// 		delete pack_string;
// 		return NULL;
// 	}
// 
// 	return pack_string;
// }
// 
